package github.sf.fw.hbase;

import github.sf.fw.tools.db.DSProperty;
import github.sf.fw.tools.db.TableDataReader;
import github.sf.fw.tools.db.model.ResultRow;
import github.sf.fw.tools.db.model.TableDataWithPage;
import github.sf.fw.tools.db.model.TableInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;

import java.sql.SQLException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class HbaseDataLoader implements TableDataReader {
    private final static Logger logger = LoggerFactory.getLogger(HbaseDataLoader.class);
    HbaseProperties hbaseProperties;
    HbaseKit hbaseKit;

    public void setDatabaseProperties(DSProperty dataSourceProps) {
        hbaseProperties = new HbaseProperties();
        BeanUtils.copyProperties(dataSourceProps, hbaseProperties);
        getHbaseKit();
    }

    public HbaseKit getHbaseKit() {
        if (hbaseKit == null) {
            hbaseKit = new HbaseKit(hbaseProperties);
        }
        return hbaseKit;
    }

    @Override
    public TableDataWithPage loadTableData(TableInfo tableInfo, int page, int size) throws SQLException {
//        Scan scan = new Scan();
//        scan.setRowOffsetPerColumnFamily((page - 1) * size);
//        scan.setMaxResultSize(size);
//        PageFilter pageFilter = new PageFilter(size);
//        Map<String, List<String>> stringListMap = hbaseKit.queryTableWithPage(tableInfo.getName(), page, size);
        Map<String, List<String>> stringListMap = null;
//        long l = hbaseKit.countTableRow(tableInfo.getName());
        long l = 0;
        try {
            CompletableFuture<Map<String, List<String>>> loadData = CompletableFuture
                .supplyAsync(() -> hbaseKit.queryTableWithPage(tableInfo.getName(), page, size));
            CompletableFuture<Long> countRows = CompletableFuture.supplyAsync(() -> hbaseKit.countTableRow(tableInfo.getName()));
            CompletableFuture.allOf(loadData, countRows).join();
            stringListMap = loadData.get();
            l = countRows.get();
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
        List<ResultRow> resultRows = HBaseUtils.convertToResultRow(stringListMap);
        return new TableDataWithPage(page, size, Math.toIntExact(l), resultRows);
    }


    @Override
    public List<ResultRow> loadTableData(TableInfo tableInfo) throws SQLException {
        return null;
    }

    @Override
    public TableDataWithPage loadTableDataWithScript(String queryScript, int page, int size) throws SQLException {
        return null;
    }

    @Override
    public List<ResultRow> loadTableDataWithScript(String querySql) throws SQLException {
        return null;
    }
}
